package com.atguigu.flink.chapter02_DataStreamAPI.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

/**
 * Created by Smexy on 2022/10/21
 *
 *      核心类: KafkaSource
 *
 *      FlinkKafkaConsumer: 从kafka读取数据。 老的API
 *      1.14开始，过时。
 *
 *
 *      推荐用的新的:  Kafka Source
 *
 *  如果方法在返回值的位置声明了泛型，此时在调用这个方法时，需要
 *          KafkaSource.<String>builder()
 *      static <OUT> KafkaSourceBuilder<OUT> builder()
 *
 *
 *      一个并行度就是一个消费者，一个消费者，可以消费一个主题的N个分区。
 *
 *  --------------------------------------
 *      flink的kafkaSource默认是把 消费的 offsets提交到当前Task的状态中维护。并不会主动提交到Kafka的_consumer_offsets中。
 *
 *      状态需要开启Checkpoint才能进行存储，后续会讲。
 *
 *      可以手动设置，顺便把Offsets提交到kafka一份。
 *
 *  -------------------------------
 *      Job重启时，如果开启了Checkpoint，默认从Checkpoint获取之前提交的offsets，获取不到时，才会从kafka的_consumer_offsets中获取。
 *
 */
public class Demo2_Kafka
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //组件，读取指定kafka主题的数据
        KafkaSource<String> source = KafkaSource.<String>builder()
            .setBootstrapServers("hadoop102:9092,hadoop103:9092")  //集群地址
            .setTopics("topicA")   //设置消费的主题
            .setGroupId("my-group") //设置消费者组id
            /*
                    从指定的位置消费：  OffsetsInitializer.offsets(Map<TopicPartition, Long> offsets)
                    从最新位置消费:    OffsetsInitializer.latest()
                    从最早位置消费:     OffsetsInitializer.earliest()
                    从上次提交的位置消费:  OffsetsInitializer.committedOffsets()
                        新的组，从来没有提交过:  OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST)
             */
            .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))   //设置起始偏移量
            //设置value的反序列化器
            .setValueOnlyDeserializer(new SimpleStringSchema())   //消费者必须设置 Key-Value的反序列化器
            //设置额外的消费者参数
            .setProperty("enable.auto.commit","true")   //允许consumer自动提交offsets
            .setProperty("auto.commit.interval.ms","1000") //每次自动提交的间隔
            .build();


            //用刚刚的那个组件去获取Source
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "hahaha")
           .print();


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
